package test.others;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.commons.collections4.CollectionUtils;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.Setter;

public class ConsistentHash {
	
	TreeMap<Integer, String> circle = new TreeMap<>();
	TreeSet<Integer> serverSet = new TreeSet<>();
	
	Map<Node, Node> virtualMap = new HashMap<>();
	Map<Node, List<Node>> virtualListMap = new HashMap<>();
	Map<Node, TreeMap<Integer, Entry>> nodeMap = new TreeMap<>();
	
	public void printCircle() {
		System.out.println("server=" + serverSet);
		int firstServerHash = serverSet.first();
		int lastServerHash = serverSet.last();
		SortedMap<Integer, String> firstBeforeMap = circle.headMap(firstServerHash);
		SortedMap<Integer, String> afterBeforeMap = circle.tailMap(lastServerHash);
		
		NavigableSet<Integer> ss = serverSet;
		while(true) {
			Integer hash1 = ss.first();
			ss = ss.tailSet(hash1, false);
			if(CollectionUtils.isEmpty(ss)) {
				System.out.println("server[hash=" + hash1 + ",ip=" +  circle.get(hash1) + "]\n\tkeys1" + firstBeforeMap + "\n\tkeys2" + afterBeforeMap);
				break;
			}
			Integer hash2 = ss.first();
			System.out.println("server[hash=" + hash1 + ",ip=" + circle.get(hash1) + "]\n\tkeys" + circle.subMap(hash1, false, hash2, false));
		}
		System.out.println("------------------------------------------");
	}
	
	
	int virtual = 0;
	
	public ConsistentHash(int virtual) {
		this.virtual = virtual;
	}
	
	public ConsistentHash() {
		
	}
	
	public void addServer(String ip) {
		int hash = hash(ip);
		circle.put(hash, ip);
		serverSet.add(hash);
		Node node = new Node(hash, ip);
		nodeMap.put(node, new TreeMap<>());
		virtualListMap.put(node, new ArrayList<>());
		//节点迁移
		transferAdd(node, hash);
		
		for(int i=0;i<virtual;i++) {
			String virtualIp = ip + "_" + i;
			int virtualHash = hash(virtualIp);
			circle.put(virtualHash, virtualIp);
			serverSet.add(virtualHash);
			Node virtualNode = new Node(virtualHash, virtualIp);
			virtualMap.put(virtualNode, node);
			virtualListMap.get(node).add(virtualNode);
			transferAdd(virtualNode, hash);
		}
		
	}

	private void transferAdd(Node node, int rootHash) {
		//节点迁移
		int hash = node.hash;
		TreeMap<Integer, String> transferMap = new TreeMap<>();
		NavigableSet<Integer> afterSet = serverSet.tailSet(hash, false);
		if(CollectionUtils.isEmpty(afterSet)) {
			afterSet = serverSet.headSet(hash, false);
			if(CollectionUtils.isEmpty(afterSet)) {
				return;//这是第一个节点不用处理
			}
			int afterHash = afterSet.first();
			transferMap.putAll(circle.headMap(afterHash, false));
			transferMap.putAll(circle.tailMap(hash, false));
		} else {
			int afterHash = afterSet.first();
			transferMap.putAll(circle.subMap(hash, false, afterHash, false));
		}
		
		NavigableSet<Integer> beforeSet = serverSet.headSet(hash, false);
		if(CollectionUtils.isEmpty(beforeSet)) {
			beforeSet = serverSet.tailSet(hash, false);
		}
		int beforeHash = beforeSet.last();
		
		if(rootHash == beforeHash) {
			return; //这种情况root迁移包含虚拟节点迁移，pass
		}
		//需要迁移的key  hash 到  afterHash
		Set<Integer> transferKeys = transferMap.keySet();
		
		Node beforeNode = new Node(beforeHash, circle.get(beforeHash));
		//虚拟节点没有存储key的map，需要转换到他的root的map
		
		if(virtualMap.containsKey(node)) {
			node = virtualMap.get(node);
		}
		
		if(virtualMap.containsKey(beforeNode)) {
			beforeNode = virtualMap.get(beforeNode);
		}
		
		for(Integer key : transferKeys) {
			
			TreeMap<Integer, Entry> tm = nodeMap.get(beforeNode);
			
			nodeMap.get(node).put(key, tm.get(key));
			tm.remove(key);
		}
	}
	
	public void deleteServer(String ip) {
		int hash = hash(ip);
		Node node = new Node(hash, ip);
		
		transferDelete(hash, node);
		
		circle.remove(hash);
		serverSet.remove(hash);
		
		
		List<Node> virtualList = virtualListMap.get(node);
		for(Node virtualNode : virtualList) {
			int virtualHash = virtualNode.hash;
			transferDelete(virtualHash, virtualNode);
			circle.remove(virtualHash);
			serverSet.remove(virtualHash);
			
			virtualMap.remove(virtualNode);
		}
		
		nodeMap.remove(node);
		virtualListMap.remove(node);
		
	}

	private void transferDelete(int hash, Node node) {
		//节点迁移
		TreeMap<Integer, String> transferMap = new TreeMap<>();
		NavigableSet<Integer> afterSet = serverSet.tailSet(hash, false);
		if(CollectionUtils.isEmpty(afterSet)) {
			afterSet = serverSet.headSet(hash, false);
			int afterHash = afterSet.first();
			transferMap.putAll(circle.headMap(afterHash, false));
			transferMap.putAll(circle.tailMap(hash, false));
		} else {
			int afterHash = afterSet.first();
			transferMap.putAll(circle.subMap(hash, false, afterHash, false));
		}
		
		NavigableSet<Integer> beforeSet = serverSet.headSet(hash, false);
		if(CollectionUtils.isEmpty(beforeSet)) {
			beforeSet = serverSet.tailSet(hash, false);
		}
		int beforeHash = beforeSet.last();
		
		//beforeHash + = (hash, afterHash)
		
		String beforeIp = circle.get(beforeHash);
		Node beforeNode = new Node(beforeHash, beforeIp);
		if(virtualMap.containsKey(beforeNode)) {
			beforeNode = virtualMap.get(beforeNode);
		}
		if(virtualMap.containsKey(node)) {
			node = virtualMap.get(node);
		}
		for(Map.Entry<Integer, String> entry : transferMap.entrySet()) {
			int transferHash = entry.getKey();
			TreeMap<Integer, Entry> tm = nodeMap.get(node);
			nodeMap.get(beforeNode).put(transferHash, tm.get(transferHash));
			tm.remove(transferHash);
		}
	}
	
	public void add(String key, String value) {
		int hash = hash(key);
		circle.put(hash, key);
		
		int serverHash = getServerHash(hash);
		String serverIp = circle.get(serverHash);
		nodeMap.get(new Node(serverHash, serverIp)).put(hash, new Entry(hash, key, value));
	}
	
	public String get(String key) {
		int hash = hash(key);
		int serverHash = getServerHash(hash);
		String serverIp = circle.get(serverHash);
		TreeMap<Integer, Entry> entryMap = nodeMap.get(new Node(serverHash, serverIp));
		String value = entryMap.get(hash).value;
		return value;
	}
	
	private int getServerHash(int hash) {
		SortedSet<Integer> resultSet = serverSet.tailSet(hash);
		if(CollectionUtils.isEmpty(resultSet)) {
			resultSet = serverSet.headSet(hash);
		} 
		int serverHash = resultSet.first();
		String ip = circle.get(serverHash);
		Node node = new Node(serverHash, ip);
		
		if(virtualMap.containsKey(node)) {
			node = virtualMap.get(node);
		}
		return node.hash;	
	}
	
	private int hash(Object key) {
		if(key==null) {
			throw new RuntimeException("key不能为空");
		}
		ByteBuffer buf = ByteBuffer.wrap(key.toString().getBytes());
		int seed = 0x1234ABCD;

		ByteOrder byteOrder = buf.order();
		buf.order(ByteOrder.LITTLE_ENDIAN);

		long m = 0xc6a4a7935bd1e995L;
		int r = 47;

		long h = seed ^ (buf.remaining() * m);

		long k;
		while (buf.remaining() >= 8) {
			k = buf.getLong();

			k *= m;
			k ^= k >>> r;
			k *= m;

			h ^= k;
			h *= m;
		}

		if (buf.remaining() > 0) {
			ByteBuffer finish = ByteBuffer.allocate(8).order(
					ByteOrder.LITTLE_ENDIAN);
			finish.put(buf).rewind();
			h ^= finish.getLong();
			h *= m;
		}

		h ^= h >>> r;
		h *= m;
		h ^= h >>> r;
		buf.order(byteOrder);
		return (int) h;
	}
	
	public static void main(String[] args) {
//		TreeSet<Integer> set = new TreeSet<>();
//		set.add(6);
//		set.add(16);
//		set.add(96);
//		set.add(36);
//		set.add(76);
//		set.add(26);
//		
//		System.out.println(set.tailSet(10));
//		System.out.println(set.tailSet(96));
//		System.out.println(set.tailSet(100));
		
		ConsistentHash ch = new ConsistentHash(2);
		
		ch.addServer("10.167.202.60");
		ch.addServer("10.167.202.86");
		ch.addServer("10.167.202.98");
		
		ch.add("hello", "world");
		ch.add("pk", "123");
		ch.add("lol", "bbq");
		ch.add("xiaoming", "name");
		ch.add("kkk", "haha");
		ch.add("biu", "hong");
		ch.add("blue", "heihei");
		ch.add("fuck", "xxx");
		ch.add("derf", "666");
		ch.add("heij", "hyiu");
		ch.add("liqi", "5454");
		ch.add("wwwq", "baidu");
		ch.add("360buy", "jd");
		ch.add("itougu", "zqt");
		ch.add("disappear", "diamond");
		ch.add("baolie", "2323");
		ch.add("13838383", "yyy");
		ch.add("8902384_888", "lis");
		ch.add("kate", "kyte");
		ch.add("amount1", "23");
		ch.add("amount2", "444");
		
		ch.printCircle();
		
		ch.addServer("10.167.202.193");
		ch.addServer("10.167.202.188");
		ch.addServer("10.167.202.194");
		
		ch.printCircle();
		
		ch.deleteServer("10.167.202.193");
		ch.deleteServer("10.167.202.188");
		ch.deleteServer("10.167.202.194");
		
		ch.printCircle();
		
	}
	
	@AllArgsConstructor
	@Data
	static class Entry {
		private int hash;
		private String key;
		private String value;
	}
	
	@AllArgsConstructor
	@Getter
	@Setter
	static class Node implements Comparable<Node>{
		private int hash;
		private String ip;
		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + hash;
			result = prime * result + ((ip == null) ? 0 : ip.hashCode());
			return result;
		}
		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			Node other = (Node) obj;
			if (hash != other.hash)
				return false;
			if (ip == null) {
				if (other.ip != null)
					return false;
			} else if (!ip.equals(other.ip))
				return false;
			return true;
		}
		@Override
		public int compareTo(Node o) {
			return this.hash - o.hash;
		}
		@Override
		public String toString() {
			return "Node [hash=" + hash + ", ip=" + ip + "]";
		}
	}

}
